package com.investmenttechnologyservices.util;

import java.io.IOException;
import java.util.List;

import com.google.appengine.tools.mapreduce.KeyValue;
import com.google.appengine.tools.mapreduce.MapReduceJob;
import com.google.appengine.tools.mapreduce.MapReduceSettings;
import com.google.appengine.tools.mapreduce.MapReduceSpecification;
import com.google.appengine.tools.mapreduce.Marshallers;
import com.google.appengine.tools.mapreduce.Output;
import com.google.appengine.tools.mapreduce.OutputWriter;
import com.google.appengine.tools.mapreduce.inputs.DatastoreInput;
import com.google.appengine.tools.mapreduce.outputs.InMemoryOutput;

public class MapReduceUtils {
	
	/**
	 * immitates SELECT SUM(sumOf) FROM kind GROUP BY sumBy
	 * 
	 * @param kind  entities to scan
	 * @param sumOf  the property to sum
	 * @param sumBy  group by property
	 * @param mapShardCount  how many mappers to spawn
	 * @param reduceShardCount  how many reducers to spawn 
	 * @return
	 */
	public static String sum(String kind, String sumOf, String sumBy, int mapShardCount, int reduceShardCount) {
		InMemoryOutput<KeyValue<String, Double>> o = new InMemoryOutput<KeyValue<String, Double>>(reduceShardCount);
		String ret = MapReduceJob.start(MapReduceSpecification.of(
				"Sum with MapReduce", 
				new DatastoreInput(kind, mapShardCount), 
				new SumMapper(sumOf, sumBy), 
				Marshallers.getStringMarshaller(),
				new DoubleMarshaller(), 
				new SumReducer(),
				o),
				getSettings());
		return ret;
	}

	private static final MapReduceSettings getSettings() {
		MapReduceSettings settings = new MapReduceSettings()
				.setWorkerQueueName("mapreduce-workers")
				.setControllerQueueName("default");
		return settings;
	}
	
	
	public static class MyOutput<O> extends Output<O, List<List<O>>> {
		private static final long serialVersionUID = 9009145518521835507L;

		@Override
		public List<? extends OutputWriter<O>> createWriters()
				throws IOException {
			return null;
		}

		@Override
		public List<List<O>> finish(List<? extends OutputWriter<O>> writers)
				throws IOException {
			return null;
		}
		
	}

}
